package com.heroh.kettle.util;

import cn.hutool.core.collection.CollectionUtil;
import com.heroh.kettle.constant.KettleConstant;
import com.heroh.kettle.entity.DbConfig;
import com.heroh.kettle.entity.ParamDbConnection;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.trans.HasDatabasesInterface;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class KettleCommonUtil {

    public static void setDatabases(HasDatabasesInterface metaI, List<ParamDbConnection> conns) {
        if (CollectionUtil.isEmpty(conns)) {
            return;
        }
        List<DatabaseMeta> databases = new ArrayList<>();
        conns.stream().forEach(conn -> {
            DatabaseMeta meta = new DatabaseMeta();
            setDabaseMeta(meta, conn);
            databases.add(meta);
        });
        metaI.setDatabases(databases);
    }

    public static void updateDatabases(HasDatabasesInterface metaI, List<ParamDbConnection> conns, Map<String, String> dbMap) {
        if (CollectionUtil.isEmpty(conns) || dbMap.isEmpty()) {
            return;
        }
        // 数据库名-数据库连接参数
        Map<String, ParamDbConnection> connsMap = conns.stream().collect(Collectors.toMap(e -> e.getDbConnectionCode(), e -> e));
        List<DatabaseMeta> existDB = metaI.getDatabases();
        for (DatabaseMeta meta : existDB) {
            // 根据meta中的别名 = dbMap.key -> dbMap.value = connsMap.key
            String dbName = dbMap.get(meta.getName());
            if (!dbMap.containsKey(meta.getName()) || !connsMap.containsKey(dbName)) {
                return;
            }
            // 数据库真实名称
            ParamDbConnection conn = connsMap.get(dbName);
            setDabaseMeta(meta, conn);
        }
        DatabaseMeta logMeta = new DatabaseMeta();
        setDabaseMeta(logMeta, connsMap.get(KettleConstant.KETTLE_LOG_DB));
        existDB.add(logMeta);
    }

    private static void setDabaseMeta(DatabaseMeta meta, ParamDbConnection conn){
        meta.setName(conn.getDbConnectionCode());
        DbConfig dbConfig = conn.getDbConfig();
        meta.setDatabaseType(dbConfig.getDbType());
        meta.setHostname(dbConfig.getIp());
        meta.setDBPort(dbConfig.getPort());
        meta.setUsername(dbConfig.getUsername());
        meta.setPassword(dbConfig.getPassword());
        meta.setDBName(dbConfig.getDbName());
        meta.addExtraOption(dbConfig.getDbType(), "characterEncoding", "utf-8");
        meta.addExtraOption(dbConfig.getDbType(), "useSSL", "false");
        if("srcdb".equals(conn.getDbConnectionCode())){
            meta.addExtraOption(dbConfig.getDbType(), "defaultFetchSize", "10000");
            meta.addExtraOption(dbConfig.getDbType(), "useServerPrepStmts", "true");
            meta.addExtraOption(dbConfig.getDbType(), "useCursorFetch", "true");
            meta.addExtraOption(dbConfig.getDbType(), "cachePrepStmts", "true");
        } else {
            meta.addExtraOption(dbConfig.getDbType(), "defaultFetchSize", "5000");
            meta.addExtraOption(dbConfig.getDbType(), "rewriteBatchedStatements", "true");
            meta.addExtraOption(dbConfig.getDbType(), "useServerPrepStmts", "false");
            meta.addExtraOption(dbConfig.getDbType(), "useCursorFetch", "true");
        }
        meta.addOptions();
    }
}
